package com.ice.work.moment.rocket.consumer;

import com.alibaba.fastjson.JSONObject;
import com.honeybee.cloud.framework.rocket.consumer.AbstractPushConsumer;
import com.ice.work.moment.facade.rocket.common.ConsumerGroupName;
import com.ice.work.moment.facade.rocket.common.TagName;
import com.ice.work.moment.facade.rocket.common.TopicName;
import com.ice.work.moment.facade.rocket.message.ImgMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * @Description
 * @Author DELL
 * @Create 2019-04-03
 * @Since 1.0.0
 */
@Slf4j
@Component
public class ImgRocketPushConsumer extends AbstractPushConsumer {

    @Resource
    private RocketMQProperties rocketMQProperties;

    @Override
    public void afterPropertiesSet() throws Exception {
        init(rocketMQProperties.getNameServer(), TopicName.IMG_TOPIC, TagName.IMG_TAG, ConsumerGroupName.CONSUMER_IMG_GROUP);
        startConsumerGroup();
    }

    @Override
    protected MessageListener listener() {
        return new ImgMessageListener();
    }
}

@Slf4j
class ImgMessageListener implements MessageListenerOrderly {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeConcurrentlyContext) {
        for (MessageExt messageExt : list) {
            try {
                String message = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                log.info("验证码消息:{}", message);
                ImgMessage imgMessage = JSONObject.parseObject(message, ImgMessage.class);

            } catch (Exception e) {
                log.error("消息处理异常", e);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
